package org.redis.cluster.statistics;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.ArrayUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.interceptor.Interceptor;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.RedisReply;
import org.redis.cluster.redis.transport.reply.RedisByteReply;

public class KeySpaceStatistics implements Plugin,Interceptor {
	
	private final static String COMMAND_SPLITOR=",";
	
	private final static String EMPTY_COMMAND_CONFIG="key_space_empty_commands";
	
	private final static String NULL_COMMAND_CONFIG="key_space_null_commands";
	
	private final static String ZERO_COMMAND_CONFIG="key_space_zero_commands";
	
	private final static String NOT_EXIST_COMMAND_CONFIG="key_space_not_exist_commands";
	
	private final static String OK_COMMAND_CONFIG="key_space_ok_commands";
	
	private final static String NONE_COMMAND_CONFIG="key_space_none_commands";
	
	private final static byte EMPTY_RESPONSE[]="*0\r\n".getBytes();
	
	private final static byte NULL_RESPONSE[]="$-1\r\n".getBytes();
	
	private final static byte ZERO_RESPONSE[]=":0\r\n".getBytes();
	
	private final static byte NOT_EXIST_RESPONSE[]=":-1\r\n".getBytes();
	
	private final static byte OK_RESPONSE[]="+OK\r\n".getBytes();
	
	private final static byte NONE_RESPONSE[]="+none\r\n".getBytes();
	
	private final static String DEFAULT_EMPTY_COMMANDS="";
	
	private final static String DEFAULT_NULL_COMMANDS="";
	
	private final static String DEFAULT_ZERO_COMMANDS="DEL,EXPIRE,EXPIREAT,PERSIST,PEXPIRE,PEXPIREAT,RENAMENX,APPEND,BITCOUNT,GETBIT,SETNX";
	
	private final static String DEFAULT_NOT_EXIST_COMMANDS="";
	
	private final static String DEFAULT_OK_COMMANDS="RENAME";
	
	private final static String DEFAULT_NONE_COMMANDS="TYPE";
	
	private final static AtomicInteger hitCounter=new AtomicInteger();
	private final static AtomicInteger missCounter=new AtomicInteger();
	
	private RedisCluster redisCluster;
	
	private Collection<String> emptyCommands;
	private Collection<String> nullCommands;
	private Collection<String> zeroCommands;
	private Collection<String> notExistCommands;
	private Collection<String> okCommands;
	private Collection<String> noneCommands;

	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.emptyCommands=initEmptyCommands();
		this.nullCommands=initNullCommands();
		this.zeroCommands=initZeroCommands();
		this.notExistCommands=initNotExistCommands();
		this.okCommands=initOkCommands();
		this.noneCommands=initNoneCommands();
	}
	
	@Override
	public void destroy() {
	}
	
	protected Collection<String> initEmptyCommands() {
		return initCommands(EMPTY_COMMAND_CONFIG, DEFAULT_EMPTY_COMMANDS);
	}
	
	protected Collection<String> initNullCommands() {
		return initCommands(NULL_COMMAND_CONFIG, DEFAULT_NULL_COMMANDS);
	}
	
	protected Collection<String> initZeroCommands() {
		return initCommands(ZERO_COMMAND_CONFIG, DEFAULT_ZERO_COMMANDS);
	}
	
	protected Collection<String> initNotExistCommands() {
		return initCommands(NOT_EXIST_COMMAND_CONFIG, DEFAULT_NOT_EXIST_COMMANDS);
	}
	
	protected Collection<String> initOkCommands() {
		return initCommands(OK_COMMAND_CONFIG, DEFAULT_OK_COMMANDS);
	}
	
	protected Collection<String> initNoneCommands() {
		return initCommands(NONE_COMMAND_CONFIG, DEFAULT_NONE_COMMANDS);
	}
	
	protected Collection<String> initCommands(String config,String defaultValue) {
		String commands=redisCluster.getConfigManager().getSystemConfig().get(config).getValue(String.class,defaultValue);
		String splitedCommands[]=commands.split(COMMAND_SPLITOR);
		if(ArrayUtils.isEmpty(splitedCommands)) {
			return Collections.emptySet();
		}
		
		return new HashSet<String>(Arrays.asList(splitedCommands));
	}

	@Override
	public RedisReply intercept(Invocation invocation) throws Throwable {
		RedisCommand command=invocation.getCommand();
		if(command.getParameterCount()<2 || !isSupport(command)) {
			return invocation.invoke();
		}
		
		RedisReply reply=invocation.invoke();
		if(!(reply instanceof RedisByteReply)) {
			return reply;
		}
		
		if(isHit(command, (RedisByteReply)reply)) {
			hit(command);
		}
		else {
			miss(command);
		}
		
		return reply;
	}
	
	protected boolean isSupport(RedisCommand command) {
		String commandStr=command.getCommand();
		return emptyCommands.contains(commandStr) || nullCommands.contains(commandStr) || zeroCommands.contains(commandStr) || notExistCommands.contains(commandStr)
				 || okCommands.contains(commandStr) || noneCommands.contains(commandStr);
	}
	
	protected boolean isHit(RedisCommand command,RedisByteReply reply) {
		byte replyValues[]=((RedisByteReply) reply).getValue();
		String commandStr=command.getCommand();
		
		if(emptyCommands.contains(commandStr)) {
			return !Arrays.equals(replyValues, EMPTY_RESPONSE);
		}
		else if(nullCommands.contains(commandStr)) {
			return !Arrays.equals(replyValues, NULL_RESPONSE);
		}
		else if(zeroCommands.contains(commandStr)) {
			return !Arrays.equals(replyValues, ZERO_RESPONSE);
		}
		else if(notExistCommands.contains(commandStr)) {
			return !Arrays.equals(replyValues, NOT_EXIST_RESPONSE);
		}
		else if(okCommands.contains(commandStr)) {
			return Arrays.equals(replyValues, OK_RESPONSE);
		}
		else if(noneCommands.contains(commandStr)) {
			return Arrays.equals(replyValues, NONE_RESPONSE);
		}
		
		return false;
	}
	
	protected void hit(RedisCommand command) {
		hitCounter.incrementAndGet();
	}
	
	protected void miss(RedisCommand command) {
		missCounter.incrementAndGet();
	}

	public static int getHitCount() {
		return hitCounter.intValue();
	}
	
	public static int getMissCount() {
		return missCounter.intValue();
	}

}